package com.shujia.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.junit.{Before, Test}

class Demo25StudentExam {
  // 定义全局变量
  var stu: RDD[Students] = _
  var sco: RDD[Score] = _
  var sub: RDD[Subject] = _

  var stuMap: Map[String, String] = _

  var subMap: Map[String, String] = _
  var subFullScoreMap: Map[String, Int] = _


  var sc: SparkContext = _

  def initSparkContext(): Unit = {
    sc = new SparkContext(new SparkConf().setAppName("Demo25StudentExam").setMaster("local[*]"))
  }

  @Before
  // 加载学生、分数、科目数据
  def loadData(): Unit = {
    initSparkContext()

    // 将学生数据中的每一条数据变成Students样例类对象 并构建一个RDD
    stu = sc
      .textFile("data/students.txt")
      .map(line => {
        val stuSplits: Array[String] = line.split(",")
        Students(stuSplits(0), stuSplits(1), stuSplits(2), stuSplits(3), stuSplits(4))
      })

    sco = sc
      .textFile("data/score.txt")
      .map(line => {
        val scoSplits: Array[String] = line.split(",")
        Score(scoSplits(0), scoSplits(1), scoSplits(2).toInt)
      })

    sub = sc
      .textFile("data/subject.txt")
      .map(line => {
        val subSplits: Array[String] = line.split(",")
        Subject(subSplits(0), subSplits(1), subSplits(2).toInt)
      })

    // 构建学生的Map 以id作为Key，name+clazz作为Value
    stuMap = stu.map(s => (s.id, s.name + "," + s.clazz)).collect().toMap

    // 构建科目的Map 以subject_id作为Key，subject_name作为Value
    subMap = sub.map(s => (s.subject_id, s.subject_name)).collect().toMap

    // 构建科目的Map（没有科目名，但有科目的总分）
    subFullScoreMap = sub.map(s => (s.subject_id, s.subject_score)).collect().toMap

  }

  def filterWithIdsAndPrint(ids: List[String]): Unit = {
    val idsBro: Broadcast[List[String]] = sc.broadcast(ids)
    val stuMapBro: Broadcast[Map[String, String]] = sc.broadcast(stuMap)
    val subMapBro: Broadcast[Map[String, String]] = sc.broadcast(subMap)

    // 遍历分数数据 再通过Map进行关联
    sco
      // 根据ids进行过滤
      .filter(s => idsBro.value.contains(s.id))
      .map(s => {
        val id: String = s.id
        val subject_id: String = s.subject_id
        // 通过学生id 在 stuMap取对应的name、clazz
        val nameAndClazz: String = stuMapBro.value.getOrElse(id, "")
        // 通过科目id 在 subMap取对应的科目名称
        val subject_name: String = subMapBro.value.getOrElse(subject_id, "")
        s"$id,$nameAndClazz,$subject_name,${s.score}"
      }).foreach(println)
  }

  @Test
  // 打印所有的数据
  def printAll(): Unit = {
    stu.take(10).foreach(println)
    sco.take(10).foreach(println)
    sub.take(10).foreach(println)
  }


  @Test
  // 3、统计每科都及格的学生 [学号，姓名，班级，科目，分数]
  def question03(): Unit = {
    // 将sco与sub关联 过滤出及格的科目成绩
    val subKVRDD: RDD[(String, Subject)] = sub
      .map(s => (s.subject_id, s))


    val ids: List[String] = sco
      .map(s => (s.subject_id, s))
      .join(subKVRDD)
      // 过滤出及格了的科目成绩
      .filter {
        case (subject_id: String, (score: Score, subject: Subject)) =>
          score.score.toDouble / subject.subject_score >= 0.6
      }
      // 统计每个人及格的科目数量
      .map {
        case (subject_id: String, (score: Score, subject: Subject)) =>
          (score.id, 1)
      }
      .reduceByKey(_ + _)
      .filter(kv => kv._2 == 6)
      .map(_._1)
      .collect()
      .toList

    filterWithIdsAndPrint(ids)


  }

  @Test
  // 4、统计每个班级的前三名 [学号，姓名，班级，分数]
  def question04(): Unit = {
    val stuMapBro: Broadcast[Map[String, String]] = sc.broadcast(stuMap)

    sco
      .map(s => (s.id, s.score))
      // 计算学生总分
      .reduceByKey(_ + _)
      // 关联学生信息数据 获取姓名和班级
      .map(kv => {
        val nameAndClazz: String = stuMapBro.value.getOrElse(kv._1, "")
        val arr: Array[String] = nameAndClazz.split(",")
        val name: String = arr(0)
        val clazz: String = arr(1)
        (kv._1, name, clazz, kv._2)
      })
      // 按照班级分组
      .groupBy(_._3)
      // 依次对每个班级进行处理
      .flatMap(kv => {
        // 按照总分降序 取出前三名
        kv._2.toList.sortBy(-_._4).take(3)
      })
      .map(t4 => s"${t4._1},${t4._2},${t4._3},${t4._4}")
      .foreach(println)

  }

  @Test
  // 5、统计偏科最严重的前100名学生  [学号，姓名，班级，科目，分数]
  def question05(): Unit = {
    val subFullScoreMapBro: Broadcast[Map[String, Int]] = sc.broadcast(subFullScoreMap)
    val varianceTop100Ids: List[String] = sco
      .map(s => {
        val subFullScore: Int = subFullScoreMapBro.value.getOrElse(s.subject_id, 0)
        (s.id, s.score.toDouble / subFullScore * 100)
      })
      .groupBy(_._1)
      .map(kv => {
        // 计算平均分
        val avgScore: Double = kv._2.map(_._2).sum / kv._2.size
        // 计算方差
        val variance: Double = kv._2.map(t2 => {
          Math.pow(t2._2 - avgScore, 2)
        }).sum / kv._2.size
        (kv._1, variance)
      })
      .sortBy(_._2, ascending = false)
      .take(100)
      .map(_._1)
      .toList

    filterWithIdsAndPrint(varianceTop100Ids)


  }

}

case class Students(id: String, name: String, age: String, gender: String, clazz: String)

case class Score(id: String, subject_id: String, score: Int)

case class Subject(subject_id: String, subject_name: String, subject_score: Int)
